package com.jecy.producer;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;

import java.util.Properties;

/**
 * Created by huijun on 16/9/12.
 */
public class KafkaProducer extends Thread {
    private String topic;
    public KafkaProducer(String topic) {
        this.topic = topic;
    }

    @Override
    public void run() {
        int i = 0;
        while (true) {
            Producer producer = creatProducer();
            producer.send(new KeyedMessage<Integer,String>(topic,"message"+i++));
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    private Producer creatProducer() {
        Properties properties = new Properties();
        properties.put("zookeeper.connect","127.0.0.1:2181");
        properties.put("serializer.class", StringEncoder.class.getName());
        properties.put("metadata.broker.list", "127.0.0.1:9092");
        properties.put("producer.type", "async");
        return new Producer<Integer,String>(new ProducerConfig(properties));
    }

    public static void main(String[]args) {
        new KafkaProducer("RESOURCE_UCENTER_UCENTER_MY_LOCAL").start();
    }
}
